package com.biz.primus.ms.sms.service.internal;

import com.biz.primus.model.misc.enums.ApiExceptionType;
import com.biz.primus.model.misc.enums.MessageType;
import com.biz.primus.model.misc.exception.ApiException;
import com.biz.primus.model.misc.vo.RuleVo;
import com.biz.primus.model.misc.vo.ShortMessageVo;
import com.biz.primus.ms.base.service.AbstractBaseService;
import com.biz.primus.ms.sms.event.ApiExceptionEvent;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Predicate;
import com.google.common.base.Splitter;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
import redis.clients.jedis.Tuple;

import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author yanweijin
 * @date 2017/5/31
 */
@Service
@Slf4j
public class FlowControlService extends AbstractBaseService {


    @Autowired
    private ShardedJedisPool jedisPool;


    //是否采用严格锁,会影响性能
    @Value("${cmp.flow.strict-lock}")
    private boolean strictLock = false;

    private String mobileLockKeyPrefix = "flow-ctrl:lock:";
    private String mobileRuleRecordKeyPrefix = "flow-ctrl:record:";


    private LoadingCache<String, List<RuleVo>> cache = CacheBuilder.newBuilder().expireAfterWrite(10, TimeUnit.MINUTES)
            .build(new CacheLoader<String, List<RuleVo>>() {
                @Override
                public List<RuleVo> load(String key) throws Exception {
                    RuleVo rule1 = new RuleVo();
                    rule1.setRuleKey("(limit-v/v2/30/sec)");
                    rule1.setTimeUnit(TimeUnit.SECONDS);
                    rule1.setTimeLength(30);
                    rule1.setMaxSent(1);
                    rule1.setOnlyMessageType(MessageType.validcode);
                    rule1.setDesciption("验证码短信30秒之内只能给同一个手机号发送1条短信");

                    RuleVo rule2 = new RuleVo();
                    rule2.setRuleKey("(limit-v/v2/5/hour)");
                    rule2.setTimeUnit(TimeUnit.HOURS);
                    rule2.setTimeLength(1);
                    rule2.setMaxSent(5);
                    rule2.setOnlyMessageType(MessageType.validcode);
                    rule2.setDesciption("验证码短信60分钟之内只能给同一个手机号发送不超过5条短信");

                    RuleVo rule3 = new RuleVo();
                    rule3.setRuleKey("(limit-v/v2/10/days)");
                    rule3.setTimeUnit(TimeUnit.DAYS);
                    rule3.setTimeLength(1);
                    rule3.setMaxSent(10);
                    rule3.setOnlyMessageType(MessageType.validcode);
                    rule3.setDesciption("验证码短信1天之内只能给同一个手机号发送不超过10条短信");

                    RuleVo rule4 = new RuleVo();
                    rule4.setRuleKey("(limit-n/v1/5/days)");
                    rule4.setTimeUnit(TimeUnit.DAYS);
                    rule4.setTimeLength(1);
                    rule4.setMaxSent(5);
                    rule4.setOnlyMessageType(MessageType.notice);
                    rule4.setDesciption("通知短信1天之内只能给同一个手机号发送不超过5条短信");

                    return Lists.newArrayList(rule1, rule2, rule3, rule4);
                }
            });

    public void checkFlowControl(ShortMessageVo sm) throws ApiException {
        log.info("into flow control ...mobile={}", sm.getMobile());
        //匹配规则
        List<RuleVo> rules = getRuleVos(sm);

        if (CollectionUtils.isEmpty(rules)) {
            return;
        }

        List<String> ruleKeys = Lists.transform(rules, new Function<RuleVo, String>() {
            @Override
            public String apply(RuleVo input) {
                return input.getRuleKey();
            }
        });
        long maxMillis = 0;
        for (RuleVo rule : rules) {
            maxMillis = Math.max(rule.getTimeMillis(), maxMillis);
        }
        long current = System.currentTimeMillis();
        long queryTimeBegin = current - maxMillis;

        try (ShardedJedis jedis = jedisPool.getResource()) {
            String key = mobileRuleRecordKeyPrefix + sm.getMobile();
            String zsetContent = new RuleRecord(sm.getRequestId(), ruleKeys).toString();

            try {
                lockMobile(sm.getMobile(), jedis);
            } catch (TimeoutException e) {
                //redis锁超时,不再检查rule,但是需要记录规则
                log.warn("flow-ctrl获取redis锁超时");
                jedis.zadd(key, current, zsetContent);
                return;
            }
            Set<Tuple> records = jedis.zrangeByScoreWithScores(key, queryTimeBegin, current + 1000 * 60);
            checkRule(sm, rules, records, current);
            jedis.zadd(key, current, zsetContent);
            unlockMobile(sm.getMobile(), jedis);
        }
    }

    /**
     * 用于异常发生后移除记录
     *
     * @param sm
     */
    public void removeFlowRecord(ShortMessageVo sm) {
        List<RuleVo> rules = getRuleVos(sm);
        if (CollectionUtils.isEmpty(rules)) {
            return;
        }
        List<String> ruleKeys = Lists.transform(rules, new Function<RuleVo, String>() {
            @Override
            public String apply(RuleVo input) {
                return input.getRuleKey();
            }
        });
        String key = mobileRuleRecordKeyPrefix + sm.getMobile();
        String zsetContent = new RuleRecord(sm.getRequestId(), ruleKeys).toString();
        try (ShardedJedis jedis = jedisPool.getResource()) {
            jedis.zrem(key, zsetContent);
        }
    }

    //寻找约束当前短信的规则
    private List<RuleVo> getRuleVos(final ShortMessageVo sm) {
        List<RuleVo> rules = findRulesByChannel(sm.getChannelCode());
        return Lists.newArrayList(Iterables.filter(rules, new Predicate<RuleVo>() {
            @Override
            public boolean apply(RuleVo input) {
                boolean pass = true;
                //此处添加过滤规则
                if (StringUtils.isNotBlank(input.getIncludeKeyword())) {
                    pass = StringUtils.contains(sm.getContent(), input.getIncludeKeyword());
                    if (!pass) {
                        return false;
                    }
                }
                if (input.getOnlyMessageType() != null) {
                    pass = sm.getType() == input.getOnlyMessageType();
                    if (!pass) {
                        return false;
                    }
                }
                if (ArrayUtils.isNotEmpty(input.getOnlySignature())) {
                    pass = ArrayUtils.contains(input.getOnlySignature(), sm.getSignature());
                    if (!pass) {
                        return false;
                    }
                }
                return pass;
            }
        }));
    }

    private void checkRule(ShortMessageVo sm, List<RuleVo> rules, Set<Tuple> records, long currentMs) {
        for (final RuleVo rule : rules) {
            final long begin = currentMs - rule.getTimeMillis();
            int effectiveRecordsSize = Sets.filter(records, new Predicate<Tuple>() {
                @Override
                public boolean apply(Tuple input) {
                    //记录时间要在规则约束时间之后,并且记录中要指明受本规则影响
                    return new RuleRecord(input).isMatch(begin, rule);

//					return input.getScore() > begin && input.getElement().contains(rule.getRuleKey());
                }
            }).size();
            if (rule.getMaxSent() <= effectiveRecordsSize) {
                publishEvent(new ApiExceptionEvent(this, sm, ApiExceptionType.FLOW_EXCEED_LIMIT, rule.getDesciption()));
                throw new ApiException(rule.getDesciption(), ApiExceptionType.FLOW_EXCEED_LIMIT);
            }
        }


    }

    private void lockMobile(String mobile, ShardedJedis jedis) throws TimeoutException {
        if (strictLock) {
            String key = mobileLockKeyPrefix + mobile;
            for (long wait = 0; jedis.setnx(key, "1") == 0; ) {
                wait += 30;
                if (wait > 300) {
                    throw new TimeoutException("获取流量控制锁超时");
                }
            }
            jedis.expire(key, 1);
        }
    }

    private void unlockMobile(String mobile, ShardedJedis jedis) {
        if (strictLock){
            jedis.del(mobileLockKeyPrefix + mobile);
        }
    }


    private List<RuleVo> findRulesByChannel(String channelCode) {
        return cache.getUnchecked("uniqueRules");
    }


    private class RuleRecord {

        private List<String> rules;

        private String requestId;

        private long timestamp;

        private static final String sp = "|:|:|";

        RuleRecord(Tuple ruleRecordTuple) {
            timestamp = Double.valueOf(ruleRecordTuple.getScore()).longValue();
            String[] split = StringUtils.split(ruleRecordTuple.getElement(), sp);
            switch (split.length) {
                case 2:
                    requestId = split[0];
                case 1:
                    rules = Splitter.on(",").splitToList(split[split.length - 1]);
                    break;
                default:
                    throw new IllegalArgumentException("rule record split length = " + split.length);
            }
        }

        RuleRecord(String requestId, List<String> ruleKeys) {
            this.rules = ruleKeys;
            this.requestId = requestId;
        }

        boolean isMatch(long begin, RuleVo ruleVo) {
            if (timestamp > begin) {
                if (rules.contains(ruleVo.getRuleKey())) {
                    return true;
                }
                //当前规则vo兼容某个被记录的规则
                if (ArrayUtils.isNotEmpty(ruleVo.getCompatibleRule())) {
                    //取交集,有交集则匹配到可兼容的规则
                    if (Sets.intersection(Sets.newHashSet(rules), Sets.newHashSet(ruleVo.getCompatibleRule())).size() > 0) {
                        return true;
                    }
                }
            }
            return false;
        }

        @Override
        public String toString() {
            return Objects.toString(requestId) + sp + Joiner.on(",").join(rules);
        }
    }


}
